package com.chukun.im.server.application.consumer;

import cn.hutool.core.util.StrUtil;
import com.chukun.im.server.application.netty.processor.MessageProcessor;
import com.chukun.im.server.application.netty.processor.factory.ProcessorFactory;
import com.chukun.im.server.common.domain.constants.IMConstants;
import com.chukun.im.server.common.domain.enums.IMCmdType;
import com.chukun.im.server.common.domain.model.IMReceiveInfo;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

/**
 * @author chukun
 * @version 1.0.0
 * @description 群聊消息消费者
 * @createTime 2024/5/5 16:40
 */
@Component
@ConditionalOnProperty(name = "message.mq.type", havingValue = "rocketmq")
@RocketMQMessageListener(consumerGroup = IMConstants.IM_MESSAGE_GROUP_CONSUMER_GROUP, topic = IMConstants.IM_MESSAGE_GROUP_EMPTY_QUEUE)
public class GroupMessageConsumer extends BaseMessageConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    private final Logger logger = LoggerFactory.getLogger(GroupMessageConsumer.class);

    @Value("${server.id}")
    private Long serverId;


    @Override
    public void onMessage(String message) {

        if (StringUtils.isEmpty(message)){
            logger.warn("groupMessageConsumer.onMessage.message.empty");
            return;
        }
        IMReceiveInfo imReceiveInfo = this.getReceiveMessage(message);
        if (imReceiveInfo == null){
            logger.warn("groupMessageConsumer.onMessage.transform.message.empty");
            return;
        }
        MessageProcessor processor = ProcessorFactory.getProcessor(IMCmdType.GROUP_MESSAGE);
        if (processor != null) {
            logger.info("groupMessageConsumer.onMessage.processor:{}", processor.getClass().getSimpleName());
            processor.process(imReceiveInfo);
        }



    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        try{
            String topic = String.join(IMConstants.MESSAGE_KEY_SPLIT, IMConstants.IM_MESSAGE_GROUP_QUEUE, String.valueOf(serverId));
            consumer.subscribe(topic, "*");
        }catch (Exception e){
            logger.error("groupMessageConsumer.prepareStart.error:{}", e.getMessage());
        }

    }
}
